package com.arch.flume.sink;

import org.apache.flume.Channel;
import org.apache.flume.Context;
import org.apache.flume.Event;
import org.apache.flume.EventDeliveryException;
import org.apache.flume.Transaction;
import org.apache.flume.conf.Configurable;
import org.apache.flume.sink.AbstractSink;

/**
 * 自定义 sink
 *
 * @author pizhihui
 * @date 2020-07-06
 */
public class MySink extends AbstractSink implements Configurable {

    @Override
    public void configure(Context context) {
        // 获取配置信息

    }

    @Override
    public Status process() throws EventDeliveryException {

        Status status = null;

        Channel channel = getChannel();
        Transaction tx = channel.getTransaction();
        tx.begin();


        try {
            Event event = null;
                event = channel.take();

            if (event != null) {

            }


            tx.commit();
            status = Status.READY;
        } catch (Exception e) {
            tx.rollback();
            status = Status.BACKOFF;

        } finally {
            tx.close();
        }

        return status;
    }


}
